package org.xfs.flume.interceptor;

import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.List;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.xfs.flume.util.JSONUtils;

public class ETLIntercetpor implements Interceptor {

	@Override
	public void initialize() {

	}

	@Override
	public Event intercept(Event event) {
		//1、取数据进行校验
		byte[]body=event.getBody();
		String log=new String(body,StandardCharsets.UTF_8);
		//2、校验
		if(JSONUtils.isValidate(log)) {
			return event;
		}
		return null;
	}

	@Override
	public List<Event> intercept(List<Event> events) {
		Iterator<Event>iterator=events.iterator();
		while(iterator.hasNext()) {
			Event next=iterator.next();
			if(intercept(next)==null) {
				iterator.remove();
			}
		}
		
		return events;
	}

	
	@Override
	public void close() {

	}

	public static class Builder implements Interceptor.Builder{

		@Override
		public void configure(Context context) {
			
		}

		@Override
		public Interceptor build() {
			return new ETLIntercetpor();
		}
		
	}
}
